CREATE OR REPLACE PACKAGE table_sync_pkg AS

  -- 主同步过程
    PROCEDURE sync_table(p_table_name IN VARCHAR2);
    
    -- 批量同步多个表
    PROCEDURE sync_tables;
    
    -- 添加新同步表
    PROCEDURE add_sync_table(
        p_table_name        VARCHAR2,
        p_source_table_name VARCHAR2,
        p_source_dblink     VARCHAR2,
        p_key_columns       VARCHAR2,
        p_timestamp_column  VARCHAR2,
        p_full_sql          CLOB,
        p_batch_size        NUMBER DEFAULT 10000,
        p_parallel_degree   NUMBER DEFAULT 4,
        p_soft_delete_col   VARCHAR2 DEFAULT NULL,
        p_extra_columns     VARCHAR2 DEFAULT NULL
    );
    
    -- 启用/禁用同步表
    PROCEDURE toggle_sync_table(
        p_table_name VARCHAR2,
        p_enable CHAR DEFAULT 'Y'
    );
    
    -- 更新全量SQL
    PROCEDURE update_full_sql(
        p_table_name VARCHAR2,
        p_full_sql CLOB
    );
    
    -- 手动执行全量同步
    PROCEDURE full_sync_table(p_table_name IN VARCHAR2);
    
    -- 查看同步状态
    FUNCTION get_sync_status(p_table_name VARCHAR2) RETURN VARCHAR2;
    
    -- 获取最后同步时间
    FUNCTION get_last_sync_time(p_table_name VARCHAR2) RETURN TIMESTAMP;
    
    -- 获取同步统计信息
    FUNCTION get_sync_stats(p_table_name VARCHAR2) RETURN NUMBER;
   
END table_sync_pkg;

CREATE OR REPLACE PACKAGE BODY table_sync_pkg AS
    
    -- 自定义记录类型（含主键标识）
    TYPE config_rec IS RECORD(
        table_name        VARCHAR2(100),      -- 目标表名
        source_table_name VARCHAR2(100),      -- 源表名
        source_dblink     VARCHAR2(100),      -- 源表所在 DB Link
        key_columns       VARCHAR2(500),      -- 主键列（逗号分隔）
        timestamp_column  VARCHAR2(100),      -- 时间戳列（用于增量同步）
        full_sql          CLOB,               -- 全量同步查询 SQL
        last_sync_time    TIMESTAMP,          -- 最后一次同步时间
        batch_size        NUMBER,             -- 分批处理批次大小
        parallel_degree   NUMBER,             -- 并行度
        soft_delete_col   VARCHAR2(100),      -- 软删除标识列
        extra_columns     VARCHAR2(4000),     -- 额外需要同步的列
        enable_flag       CHAR(1),            -- 启用标志（Y/N）
        has_primary_key   BOOLEAN             -- 是否有主键（根据 key_columns 判断）
    );
    
    --日志控制参数：生产环境建议设为 FALSE 关闭日志输出
    g_enable_log BOOLEAN := TRUE;
    
    -- 列存在性缓存（避免重复查询数据字典，提升效率）
    TYPE column_cache_type IS TABLE OF BOOLEAN INDEX BY VARCHAR2(400);
    g_column_cache column_cache_type;
    
    -- 主键列缓存（缓存表主键信息，减少重复解析）
    TYPE pk_cache_type IS TABLE OF VARCHAR2(500) INDEX BY VARCHAR2(100);
    g_pk_cache pk_cache_type;
   
   -- 辅助函数：兼容多版本的行号获取
   -- 作用：从调用栈中解析出当前执行代码的行号，用于日志定位
   FUNCTION get_line_number RETURN PLS_INTEGER IS
	    v_stack VARCHAR2(4000);
	    v_line  PLS_INTEGER := 0;
        v_start_pos PLS_INTEGER := 1;
        v_end_pos PLS_INTEGER;
        v_current_line VARCHAR2(400);
        v_found BOOLEAN := FALSE;
        -- 跳过的内部辅助函数，避免日志行号干扰
        v_skip_procedures VARCHAR2(200) := 'GET_LINE_NUMBER|LOG_MSG|LOG_SQL|LOG_SQL_RESULT|LOG_ERROR';
	BEGIN
	    v_stack := DBMS_UTILITY.FORMAT_CALL_STACK;
        
        -- 跳过前两行标题（调用栈头信息）
        v_start_pos := INSTR(v_stack, CHR(10), 1, 1); -- 第一个换行
        IF v_start_pos > 0 THEN
            v_start_pos := INSTR(v_stack, CHR(10), v_start_pos+1, 1); -- 第二个换行
            IF v_start_pos > 0 THEN
                v_start_pos := v_start_pos + 1; -- 指向第三行开始
            END IF;
        END IF;
        
        -- 从第三行开始遍历调用栈,遍历调用栈，提取有效行号（筛选包含本包且非辅助函数的行）
        WHILE v_start_pos > 0 AND v_start_pos <= LENGTH(v_stack) LOOP
            v_end_pos := INSTR(v_stack, CHR(10), v_start_pos); -- 下一个换行符
            IF v_end_pos = 0 THEN
                v_end_pos := LENGTH(v_stack) + 1;
            END IF;
            
            v_current_line := SUBSTR(v_stack, v_start_pos, v_end_pos - v_start_pos);
            
            -- 正则提取行号，异常时设为 0
            BEGIN
                v_line := TO_NUMBER(REGEXP_SUBSTR(v_current_line, '\s+(\d+)\s+', 1, 1, NULL, 1));
            EXCEPTION
                WHEN OTHERS THEN
                    v_line := 0;
            END;
            
            -- 检查是否为有效行（包含包名且不在跳过列表）
            IF v_line > 0 
               AND INSTR(v_current_line, 'TABLE_SYNC_PKG') > 0
               AND NOT REGEXP_LIKE(v_current_line, v_skip_procedures, 'i') 
            THEN
                v_found := TRUE;
                EXIT;
            END IF;
            
            v_start_pos := CASE WHEN v_end_pos < LENGTH(v_stack) THEN v_end_pos + 1 ELSE 0 END;
        END LOOP;
        
        RETURN CASE WHEN v_found THEN v_line ELSE 0 END;
	END;

    -- 辅助过程：条件日志输出（根据 g_enable_log 控制是否打印）
    -- 参数：p_msg 要输出的日志内容
    PROCEDURE log_msg(p_msg VARCHAR2) IS
    	v_line PLS_INTEGER := get_line_number; 
    BEGIN
        IF g_enable_log THEN
            DBMS_OUTPUT.PUT_LINE(TO_CHAR(SYSTIMESTAMP, 'YYYY-MM-DD HH24:MI:SS.FF') 
            || ' [Line:' || v_line || ']' 
            || ' - ' || p_msg);
        END IF;
    END;
    
    -- 辅助过程：SQL 执行日志（记录执行的 SQL 语句及参数）
    -- 参数：p_sql 要执行的 SQL；p_params 可选，SQL 绑定参数
    PROCEDURE log_sql(p_sql VARCHAR2, p_params VARCHAR2 DEFAULT NULL) IS
    	v_line PLS_INTEGER := get_line_number; 
    BEGIN
        IF g_enable_log THEN
            log_msg(' 执行 SQL (行号:' || v_line || '): ' || SUBSTR(p_sql, 1, 4000));
            IF p_params IS NOT NULL THEN
                log_msg(' 参数: ' || p_params);
            END IF;
        END IF;
    END;
    
    -- 辅助过程：SQL 执行结果日志（记录影响行数、耗时）
    -- 参数：p_action 操作描述；p_rows_affected 影响行数；p_start_time 执行开始时间
    PROCEDURE log_sql_result(p_action VARCHAR2, p_rows_affected NUMBER, p_start_time TIMESTAMP) IS
        v_interval INTERVAL DAY TO SECOND;
        v_elapsed_seconds NUMBER;
        v_elapsed_time_str VARCHAR2(50);
        v_line PLS_INTEGER := get_line_number; 
    BEGIN
        IF g_enable_log THEN
            v_interval := SYSTIMESTAMP - p_start_time;
            v_elapsed_seconds := EXTRACT(DAY FROM v_interval) * 24 * 60 * 60 +
                                 EXTRACT(HOUR FROM v_interval) * 60 * 60 +
                                 EXTRACT(MINUTE FROM v_interval) * 60 +
                                 EXTRACT(SECOND FROM v_interval);
            v_elapsed_time_str := TO_CHAR(v_elapsed_seconds, '999999.999');
            log_msg('SQL 执行完成 - ' || p_action || ': 影响行数 = ' || p_rows_affected ||
                    ', 耗时 = ' || v_elapsed_time_str || ' 秒');
        END IF;
    END;
    
    -- 辅助过程：错误日志输出（突出 ERROR 标识，便于排查）
    -- 参数：p_msg 错误信息
    PROCEDURE log_error(p_msg VARCHAR2) IS
    	v_line PLS_INTEGER := get_line_number;
    BEGIN
        IF g_enable_log THEN
            DBMS_OUTPUT.PUT_LINE(TO_CHAR(SYSTIMESTAMP, 'YYYY-MM-DD HH24:MI:SS.FF') 
            || ' [Line:' || v_line || '] - ERROR: ' 
            || ' - ERROR: ' || p_msg);
        END IF;
    END;
    
    -- 辅助函数：检查列是否存在（支持 DB Link，带缓存避免重复查询）
    -- 参数：p_table_name 表名；p_column_name 列名；p_dblink 可选，DB Link
    -- 返回：列存在返回 TRUE，否则 FALSE
    FUNCTION is_column_exists(
        p_table_name  VARCHAR2,
        p_column_name VARCHAR2,
        p_dblink      VARCHAR2 DEFAULT NULL
    ) RETURN BOOLEAN IS
        v_cache_key VARCHAR2(400) := p_table_name || '.' || p_column_name || 
                                    CASE WHEN p_dblink IS NOT NULL THEN '@' || p_dblink ELSE '' END;
        v_test_sql VARCHAR2(1000);
        v_dummy NUMBER;
        v_is_remote BOOLEAN := (p_dblink IS NOT NULL);
    BEGIN
        -- 检查缓存
        IF g_column_cache.EXISTS(v_cache_key) THEN
            RETURN g_column_cache(v_cache_key);
        END IF;
        
        log_msg(' 检查列存在性: ' || p_table_name || '.' || p_column_name || 
                CASE WHEN v_is_remote THEN '@' || p_dblink ELSE '' END);
                
        -- 使用数据字典查询
        IF v_is_remote THEN
            v_test_sql := 'SELECT COUNT(*) FROM ALL_TAB_COLUMNS@' || p_dblink || 
                          ' WHERE TABLE_NAME = ''' || UPPER(p_table_name) || '''' ||
                          ' AND COLUMN_NAME = ''' || UPPER(p_column_name) || '''';
            EXECUTE IMMEDIATE v_test_sql INTO v_dummy;
        ELSE
            SELECT COUNT(*) INTO v_dummy
            FROM ALL_TAB_COLUMNS
            WHERE TABLE_NAME = UPPER(p_table_name)
            AND COLUMN_NAME = UPPER(p_column_name);
        END IF;
        
        -- 记录缓存并返回结果
        g_column_cache(v_cache_key) := (v_dummy > 0);
        RETURN (v_dummy > 0);
    EXCEPTION
        WHEN OTHERS THEN
            log_error(' 检查列存在性时出错: ' || SQLERRM);
            -- 发生错误时不缓存结果，避免误判
            RETURN FALSE;
    END;
    
    -- 构建主键条件（含详细校验，确保目标表、源表主键列存在）
    -- 参数：p_key_columns 主键列（逗号分隔）；p_prefix1 目标表别名；p_prefix2 源表别名；
    -- p_target_table 目标表名；p_source_table 源表名；p_source_dblink 源表 DB Link
    -- 返回：拼接好的主键关联条件（如 t.COL1 = s.COL1 AND t.COL2 = s.COL2 ）
    FUNCTION build_key_condition(
        p_key_columns    VARCHAR2,
        p_prefix1        VARCHAR2 DEFAULT 't',
        p_prefix2        VARCHAR2 DEFAULT 's',
        p_target_table   VARCHAR2,
        p_source_table   VARCHAR2,
        p_source_dblink  VARCHAR2
    ) RETURN VARCHAR2 IS
        v_condition VARCHAR2(4000);
        v_key VARCHAR2(100);
        v_delimiter VARCHAR2(1) := ',';
        v_start_pos PLS_INTEGER := 1;
        v_end_pos PLS_INTEGER;
        v_key_count NUMBER := 0;
    BEGIN
        log_msg(' 构建主键条件，原始列: ' || p_key_columns);
        v_condition := '';
        
        LOOP
            v_end_pos := INSTR(p_key_columns, v_delimiter, v_start_pos);
            IF v_end_pos = 0 THEN
                v_key := TRIM(SUBSTR(p_key_columns, v_start_pos));
                v_key_count := v_key_count + 1;
            ELSE
                v_key := TRIM(SUBSTR(p_key_columns, v_start_pos, v_end_pos - v_start_pos));
                v_key_count := v_key_count + 1;
            END IF;
            
            log_msg(' 处理主键列 #' || v_key_count || ': ' || v_key);
            
            -- 校验目标表列
            IF NOT is_column_exists(p_target_table, v_key) THEN
                RAISE_APPLICATION_ERROR(-20003, 
                                        ' 列 "' || v_key || '" 在目标表 ' || p_target_table || ' 中不存在');
            END IF;
            
            -- 校验源表列（支持 DB Link）
            IF NOT is_column_exists(p_source_table, v_key, p_source_dblink) THEN
                RAISE_APPLICATION_ERROR(-20003, 
                                        ' 列 "' || v_key || '" 在源表 ' || p_source_table || ' 中不存在');
            END IF;
            
            -- 构建条件
            IF v_condition IS NULL THEN
                v_condition := p_prefix1 || '.' || v_key || ' = ' || p_prefix2 || '.' || v_key;
            ELSE
                v_condition := v_condition || ' AND ' || p_prefix1 || '.' || v_key || ' = ' || p_prefix2 || '.' || v_key;
            END IF;
            
            log_msg(' 添加条件: ' || p_prefix1 || '.' || v_key || ' = ' || p_prefix2 || '.' || v_key);
            
            IF v_end_pos = 0 THEN
                EXIT;
            END IF;
            v_start_pos := v_end_pos + 1;
        END LOOP;
        
        log_msg(' 成功处理 ' || v_key_count || ' 个主键列');
        log_msg(' 最终条件: ' || v_condition);
        RETURN v_condition;
    END;
   
     -- 生成缺失列的默认值表达式（处理源表没有、目标表有的列，赋予默认值）
    -- 参数：p_table_name 目标表名；p_key_columns 主键列；p_valid_extra_cols 有效额外列；
    -- p_source_table 源表名；p_source_dblink 源表 DB Link
    -- 返回：拼接好的默认值表达式（如 0 AS IS_DELETED, 'DEFAULT_VALUE' AS LOCAL_ATTR ）
	FUNCTION generate_default_expressions(
	    p_table_name    VARCHAR2,
	    p_key_columns   VARCHAR2,
	    p_valid_extra_cols VARCHAR2,
	    p_source_table  VARCHAR2,
	    p_source_dblink VARCHAR2
	) RETURN VARCHAR2 IS
	    v_owner VARCHAR2(100);
	    v_table VARCHAR2(100);
	    v_all_cols VARCHAR2(32767);
	    v_default_exprs VARCHAR2(32767) := '';
	    v_cols VARCHAR2(32767);
	    v_idx PLS_INTEGER := 1;
	    v_col VARCHAR2(100);
	    v_pos PLS_INTEGER;
	BEGIN
	    -- 解析表名和模式
	    IF INSTR(p_table_name, '.') > 0 THEN
	        v_owner := SUBSTR(p_table_name, 1, INSTR(p_table_name, '.') - 1);
	        v_table := SUBSTR(p_table_name, INSTR(p_table_name, '.') + 1);
	    ELSE
	        v_owner := USER;
	        v_table := p_table_name;
	    END IF;
	    
	    -- 获取目标表所有列
	    SELECT LISTAGG(column_name, ',') WITHIN GROUP(ORDER BY column_id)
	    INTO v_all_cols
	    FROM all_tab_columns
	    WHERE owner = UPPER(v_owner)
	    AND table_name = UPPER(v_table)
	    AND data_type NOT IN ('CLOB', 'BLOB', 'BFILE', 'LONG', 'LONG RAW');
	    
	    log_msg(' 目标表 ' || p_table_name || ' 的所有列: ' || v_all_cols);
	    v_cols := v_all_cols;
	    
	    WHILE v_idx <= LENGTH(v_cols) LOOP
	        v_pos := INSTR(v_cols, ',', v_idx);
	        IF v_pos > 0 THEN
	            v_col := TRIM(SUBSTR(v_cols, v_idx, v_pos - v_idx));
	            v_idx := v_pos + 1;
	        ELSE
	            v_col := TRIM(SUBSTR(v_cols, v_idx));
	            v_idx := LENGTH(v_cols) + 1;
	        END IF;
	        
	        -- 跳过主键列和有效额外列
	        IF INSTR(',' || p_key_columns || ',', ',' || v_col || ',') > 0 OR
	           INSTR(',' || p_valid_extra_cols || ',', ',' || v_col || ',') > 0 THEN
	            CONTINUE;
	        END IF;
	        
	        -- 检查源表是否存在该列
	        IF NOT is_column_exists(p_source_table, v_col, p_source_dblink) THEN
	            -- 生成默认值表达式
	            v_default_exprs := v_default_exprs || CASE 
	                WHEN v_default_exprs IS NULL THEN '' ELSE ', ' END ||
	                CASE
	                    WHEN v_col = 'IS_DELETED' THEN '0 AS ' || v_col
	                   -- WHEN v_col LIKE '%_TIME' OR v_col = 'UPDATED_TIME' THEN 'SYSTIMESTAMP AS ' || v_col
	                    WHEN v_col = 'LOCAL_ATTR' THEN '''DEFAULT_VALUE'' AS ' || v_col
	                    WHEN v_col = 'VERSION' THEN '1 AS ' || v_col
	                    WHEN v_col = 'STATUS' THEN '''ACTIVE'' AS ' || v_col
	                    ELSE 'NULL AS ' || v_col  -- 其他列默认设为NULL
	                END;
	            log_msg(' 为缺失列 ' || v_col || ' 生成默认值表达式');
	        END IF;
	    END LOOP;
	    
	    RETURN v_default_exprs;
	END;
    
    -- 获取列列表（确保包含主键并处理默认值）
	FUNCTION build_column_list(
	    p_table_name    VARCHAR2,
	    p_key_columns   VARCHAR2,
	    p_extra_columns VARCHAR2,
	    p_source_table  VARCHAR2,
	    p_source_dblink VARCHAR2
	) RETURN VARCHAR2 IS
	    v_owner VARCHAR2(100);
	    v_table VARCHAR2(100);
	    v_all_cols VARCHAR2(32767);
	    v_result VARCHAR2(32767) := '';
	    v_cols VARCHAR2(32767);
	    v_start_time TIMESTAMP := SYSTIMESTAMP;
	    v_valid_extra_cols VARCHAR2(1000) := '';
	    v_default_cols VARCHAR2(32767) := ''; -- 存储需要设置默认值的列
	BEGIN
	    log_msg(' 获取表 ' || p_table_name || ' 列列表，主键列: ' || p_key_columns);
	    
	    -- 解析表名和模式
	    IF INSTR(p_table_name, '.') > 0 THEN
	        v_owner := SUBSTR(p_table_name, 1, INSTR(p_table_name, '.') - 1);
	        v_table := SUBSTR(p_table_name, INSTR(p_table_name, '.') + 1);
	    ELSE
	        v_owner := USER;
	        v_table := p_table_name;
	    END IF;
	    
	    -- 获取目标表所有列
	    SELECT LISTAGG(column_name, ',') WITHIN GROUP(ORDER BY column_id)
	    INTO v_all_cols
	    FROM all_tab_columns
	    WHERE owner = UPPER(v_owner)
	    AND table_name = UPPER(v_table)
	    AND data_type NOT IN ('CLOB', 'BLOB', 'BFILE', 'LONG', 'LONG RAW');
	    
	    log_msg(' 目标表 ' || p_table_name || ' 的所有列: ' || v_all_cols);
	    
	    -- 过滤源表存在的额外列
	    IF p_extra_columns IS NOT NULL AND TRIM(p_extra_columns) IS NOT NULL THEN
	        DECLARE
	            v_extra_col VARCHAR2(100);
	            v_extra_pos PLS_INTEGER := 1;
	            v_extra_end PLS_INTEGER;
	        BEGIN
	            LOOP
	                v_extra_end := INSTR(p_extra_columns, ',', v_extra_pos);
	                IF v_extra_end > 0 THEN
	                    v_extra_col := TRIM(SUBSTR(p_extra_columns, v_extra_pos, v_extra_end - v_extra_pos));
	                    v_extra_pos := v_extra_end + 1;
	                ELSE
	                    v_extra_col := TRIM(SUBSTR(p_extra_columns, v_extra_pos));
	                    v_extra_pos := LENGTH(p_extra_columns) + 1;
	                END IF;
	                
	                -- 检查源表是否存在该额外列（支持 DB Link）
	                IF is_column_exists(p_source_table, v_extra_col, p_source_dblink) THEN
	                    v_valid_extra_cols := v_valid_extra_cols || ',' || v_extra_col;
	                    log_msg(' 额外列 ' || v_extra_col || ' 在源表中存在，保留');
	                ELSE
	                    log_msg(' 额外列 ' || v_extra_col || ' 在源表中不存在，将设置默认值');
	                END IF;
	                
	                IF v_extra_pos > LENGTH(p_extra_columns) THEN
	                    EXIT;
	                END IF;
	            END LOOP;
	            v_valid_extra_cols := LTRIM(v_valid_extra_cols, ',');
	        END;
	    END IF;
	    
	    -- 构建列列表（排除主键列，包含有效额外列）
	    v_cols := v_all_cols;
	    v_result := '';
	    
	    DECLARE
	        v_idx PLS_INTEGER := 1;
	        v_col VARCHAR2(100);
	        v_pos PLS_INTEGER;
	    BEGIN
	        WHILE v_idx <= LENGTH(v_cols) LOOP
	            v_pos := INSTR(v_cols, ',', v_idx);
	            IF v_pos > 0 THEN
	                v_col := TRIM(SUBSTR(v_cols, v_idx, v_pos - v_idx));
	                v_idx := v_pos + 1;
	            ELSE
	                v_col := TRIM(SUBSTR(v_cols, v_idx));
	                v_idx := LENGTH(v_cols) + 1;
	            END IF;
	            
	            -- 过滤非主键列和有效额外列
	            IF INSTR(',' || p_key_columns || ',', ',' || v_col || ',') = 0 AND
	               INSTR(',' || v_valid_extra_cols || ',', ',' || v_col || ',') = 0 THEN
	                v_result := v_result || v_col || ',';
	            END IF;
	        END LOOP;
	    END;
	    
	    -- 添加主键列（确保包含主键）
	    IF p_key_columns IS NOT NULL AND TRIM(p_key_columns) IS NOT NULL THEN
	        v_result := v_result || p_key_columns || ',';
	    END IF;
	    
	    -- 添加有效额外列
	    v_result := v_result || v_valid_extra_cols;
	    
	    -- 处理源表不存在的列，生成默认值表达式
	    v_default_cols := generate_default_expressions(p_table_name, p_key_columns, v_valid_extra_cols, p_source_table, p_source_dblink);
	    IF v_default_cols IS NOT NULL AND TRIM(v_default_cols) != '' THEN
	        v_result := v_result || ',' || v_default_cols;
	    END IF;
	    
	    -- 移除最后一个逗号
	    IF v_result IS NOT NULL AND LENGTH(v_result) > 0 THEN
	        v_result := RTRIM(v_result, ',');
	    END IF;
	    
	    log_msg(' 最终列列表: ' || v_result);
	    log_sql_result(' 构建列列表 ', 0, v_start_time);
	    RETURN v_result;
	END;

    -- 生成带默认值的SELECT列列表
	FUNCTION generate_select_columns(
	    p_col_list VARCHAR2,
	    p_source_table VARCHAR2,
	    p_source_dblink VARCHAR2
	) RETURN VARCHAR2 IS
	    v_cols VARCHAR2(32767) := p_col_list;
	    v_result VARCHAR2(32767) := '';
	    v_idx PLS_INTEGER := 1;
	    v_col VARCHAR2(100);
	    v_pos PLS_INTEGER;
	BEGIN
	    WHILE v_idx <= LENGTH(v_cols) LOOP
	        v_pos := INSTR(v_cols, ',', v_idx);
	        IF v_pos > 0 THEN
	            v_col := TRIM(SUBSTR(v_cols, v_idx, v_pos - v_idx));
	            v_idx := v_pos + 1;
	        ELSE
	            v_col := TRIM(SUBSTR(v_cols, v_idx));
	            v_idx := LENGTH(v_cols) + 1;
	        END IF;
	        
	        -- 检查源表是否存在该列
	        IF is_column_exists(p_source_table, v_col, p_source_dblink) THEN
	            v_result := v_result || CASE 
	                WHEN v_result IS NULL THEN '' ELSE ', ' END || 
	                's.' || v_col;
	        ELSE
	            -- 源表不存在，使用默认值表达式（从列列表构建时已处理）
	            v_result := v_result || CASE 
	                WHEN v_result IS NULL THEN '' ELSE ', ' END || 
	                CASE
                    	WHEN v_col = 'IS_DELETED' THEN '0'
                    	WHEN v_col = 'LOCAL_ATTR' THEN '''DEFAULT_VALUE'''
                    -- 其他字段默认值规则...
                    ELSE 'NULL' 
                END; -- 列列表中已包含默认值表达式
	        END IF;
	    END LOOP;
	    
	    RETURN v_result;
	END;

    
    -- 获取表配置（含主键标识）
    FUNCTION get_config(p_table_name VARCHAR2) RETURN config_rec IS
        v_config config_rec;
        v_has_pk_num NUMBER;
    BEGIN
        log_msg(' 获取表 ' || p_table_name || ' 同步配置');
        
        SELECT table_name,
               source_table_name,
               source_dblink,
               key_columns,
               timestamp_column,
               full_sql,
               last_sync_time,
               batch_size,
               parallel_degree,
               soft_delete_col,
               extra_columns,
               enable_flag,
               CASE WHEN key_columns IS NOT NULL AND  TRIM(key_columns) IS NOT NULL THEN 1 ELSE 0 END
        INTO v_config.table_name,
             v_config.source_table_name,
             v_config.source_dblink,
             v_config.key_columns,
             v_config.timestamp_column,
             v_config.full_sql,
             v_config.last_sync_time,
             v_config.batch_size,
             v_config.parallel_degree,
             v_config.soft_delete_col,
             v_config.extra_columns,
             v_config.enable_flag,
             v_has_pk_num
        FROM sync_config
        WHERE table_name = UPPER(p_table_name)
        AND enable_flag = 'Y';
        
        v_config.has_primary_key := (v_has_pk_num = 1);
        log_msg(' 配置获取成功，源表: ' || v_config.source_table_name);
        RETURN v_config;
    EXCEPTION
        WHEN NO_DATA_FOUND THEN
            log_error(' 未找到表 ' || p_table_name || ' 的同步配置');
            RAISE_APPLICATION_ERROR(-20001, ' 表 ' || p_table_name || ' 未配置或已禁用');
    END;
    
    -- 构建更新列表（排除主键列，处理默认值）
    FUNCTION build_update_list(
        p_table_name    VARCHAR2,
        p_key_columns   VARCHAR2,
        p_extra_columns VARCHAR2,
        p_source_table  VARCHAR2,
        p_source_dblink VARCHAR2
    ) RETURN VARCHAR2 IS
        v_config config_rec;
        v_col_list VARCHAR2(32767);
        v_update_list VARCHAR2(32767) := '';
        v_col VARCHAR2(100);
        v_delimiter VARCHAR2(1) := ',';
        v_start_pos PLS_INTEGER := 1;
        v_end_pos PLS_INTEGER;
    BEGIN
        log_msg(' 构建更新列表，表: ' || p_table_name);
        
        v_config := get_config(p_table_name);
        v_col_list := build_column_list(p_table_name, p_key_columns, p_extra_columns, p_source_table, p_source_dblink);
        
        log_msg(' 非主键列列表: ' || v_col_list);
        
        v_start_pos := 1;
        LOOP
            v_end_pos := INSTR(v_col_list, v_delimiter, v_start_pos);
            v_col := TRIM(SUBSTR(v_col_list, v_start_pos, v_end_pos - v_start_pos));
            
            -- 跳过主键列
            IF INSTR(',' || p_key_columns || ',', ',' || v_col || ',') > 0 THEN
                v_start_pos := v_end_pos + 1;
                IF v_end_pos = 0 THEN
                    EXIT;
                END IF;
                CONTINUE;
            END IF;
            
            -- 构建更新项（支持 DB Link 列检查）
            IF is_column_exists(p_source_table, v_col, p_source_dblink) THEN
                v_update_list := v_update_list || CASE 
                    WHEN v_update_list IS NULL THEN '' ELSE ', ' END || 
                    't.' || v_col || ' = s.' || v_col;
                log_msg(' 更新列 ' || v_col || ' = s.' || v_col);
            ELSE
                -- 源表不存在的列，设置默认值
                IF v_col = 'IS_DELETED' THEN
                    v_update_list := v_update_list || CASE 
                        WHEN v_update_list IS NULL THEN '' ELSE ', ' END || 
                        't.' || v_col || ' = 0';
                    log_msg(' 源表无列 ' || v_col || '，设置默认值 0');
                /*ELSIF v_col LIKE '%_TIME' OR v_col = 'UPDATED_TIME' THEN
                    v_update_list := v_update_list || CASE 
                        WHEN v_update_list IS NULL THEN '' ELSE ', ' END || 
                        't.' || v_col || ' = SYSTIMESTAMP';
                    log_msg(' 源表无列 ' || v_col || '，设置默认值 SYSTIMESTAMP');*/
                ELSIF v_col = 'LOCAL_ATTR' THEN
                    v_update_list := v_update_list || CASE 
                        WHEN v_update_list IS NULL THEN '' ELSE ', ' END || 
                        't.' || v_col || ' = ''DEFAULT_VALUE''';
                    log_msg(' 源表无列 ' || v_col || '，设置默认值 DEFAULT_VALUE');
                ELSE
                    log_msg(' before: ' || v_update_list);
                                    
                    IF v_col IS NOT NULL AND TRIM(v_col) <> '' THEN
				        -- 根据业务需求，这里“保留原值”的正确拼接，比如：
				        v_update_list := v_update_list || CASE 
				            WHEN v_update_list IS NULL THEN '' ELSE ', ' END || 
				            't.' || v_col || ' = t.' || v_col;
				        log_msg(' 源表无列 ' || v_col || '，保留原值');
				    ELSE
				        -- 若v_col为空，可记录错误日志或做其他处理，避免无效拼接
				        log_error(' 列名为空，跳过无效拼接');
				    END IF;
				    log_msg(' after: ' || v_update_list);
                    log_msg(' 源表无列 ' || v_col || '，保留原值');
                END IF;
            END IF;
            
            v_start_pos := v_end_pos + 1;
            IF v_end_pos = 0 THEN
                EXIT;
            END IF;
        END LOOP;
        
        log_msg(' 最终更新列表: ' || v_update_list);
        RETURN v_update_list;
    END;
    
    -- 记录同步历史
    PROCEDURE log_sync_history(
        p_table_name    VARCHAR2,
        p_sync_type     VARCHAR2,
        p_operation     VARCHAR2,
        p_rows_affected NUMBER,
        p_status        VARCHAR2,
        p_error_message VARCHAR2 DEFAULT NULL,
        p_batch_id      NUMBER DEFAULT NULL
    ) IS
        v_history_id NUMBER;
        v_error_id NUMBER;
    BEGIN
        log_msg(' 记录同步历史：表 = ' || p_table_name || ', 操作 = ' || p_operation);
        -- 插入历史记录
        INSERT INTO sync_history
        (history_id, table_name, sync_type, operation, start_time, end_time, 
         rows_affected, status, error_message, batch_id)
        VALUES
        (sync_history_seq.NEXTVAL, p_table_name, p_sync_type, p_operation, 
         SYSTIMESTAMP, SYSTIMESTAMP, p_rows_affected, p_status, p_error_message, p_batch_id)
        RETURNING history_id INTO v_history_id;
        
        -- 记录错误日志
        IF p_status = 'ERROR' AND p_error_message IS NOT NULL THEN
            INSERT INTO sync_error_log
            (error_id, history_id, error_detail, create_time)
            VALUES
            (sync_error_log_seq.NEXTVAL, v_history_id, p_error_message, SYSTIMESTAMP)
            RETURNING error_id INTO v_error_id;
            log_msg(' 错误日志 ID: ' || v_error_id);
        END IF;
        
        COMMIT;
        log_msg(' 历史记录 ID: ' || v_history_id);
    EXCEPTION
        WHEN OTHERS THEN
            log_error(' 记录历史失败: ' || SQLERRM);
    END;
    
    -- 分批处理核心过程
    PROCEDURE process_batch(
        p_table_name    VARCHAR2,
        p_source_query  VARCHAR2,
        p_batch_size    NUMBER,
        p_key_columns   VARCHAR2,
        p_timestamp_col VARCHAR2,
        p_col_list      VARCHAR2,
        p_batch_id      NUMBER,
        p_operation     VARCHAR2,
        p_source_dblink VARCHAR2
    ) IS
        v_config config_rec;
        v_paged_query VARCHAR2(32767);
        v_start_row NUMBER := 1;
        v_end_row NUMBER := p_batch_size;
        v_total_rows NUMBER := 0;
        v_batch_count NUMBER := 0;
        v_has_pk BOOLEAN;
        v_start_time TIMESTAMP := SYSTIMESTAMP;
        v_select_cols VARCHAR2(32767);
        v_count_sql VARCHAR2(32767);
    BEGIN
        v_config := get_config(p_table_name);
        v_has_pk := v_config.has_primary_key;
        
        log_msg(' 分批处理初始化，批次大小: ' || p_batch_size);
       
        -- 生成带默认值的SELECT列列表
	    v_select_cols := generate_select_columns(p_col_list, v_config.source_table_name, v_config.source_dblink);
        
        -- 计算总记录数
		v_count_sql := 'SELECT COUNT(*) FROM (' || p_source_query || ')'; -- 定义要执行的 SQL 变量
		log_sql('process_batch 计算总记录数 :' || v_count_sql); -- 调用 log_sql 过程，记录即将执行的 SQL 语句
		EXECUTE IMMEDIATE v_count_sql INTO v_total_rows;
	
        v_batch_count := CEIL(v_total_rows / p_batch_size);
        
        log_msg(' 总记录数: ' || v_total_rows || ', 批次数: ' || v_batch_count);
        
        -- 分批处理
        FOR i IN 1 .. v_batch_count LOOP
            log_msg(' 处理批次 ' || i || '/' || v_batch_count);
            
            -- 构建分页查询
            v_paged_query := 'SELECT * FROM (
                SELECT t.*, ROWNUM rn FROM (
                    ' || p_source_query || '
                ) t WHERE ROWNUM <= ' || v_end_row || '
            ) WHERE rn >= ' || v_start_row;
            
            -- 执行 MERGE 操作
            DECLARE
                v_merge_sql VARCHAR2(32767);
                v_key_cond VARCHAR2(4000);
                v_update_list VARCHAR2(32767);
                v_time_cond VARCHAR2(1000);
            BEGIN
                v_key_cond := build_key_condition(
                    p_key_columns, 't', 's', p_table_name, 
                    v_config.source_table_name, p_source_dblink
                );
                
                v_update_list := build_update_list(
                    p_table_name, p_key_columns, v_config.extra_columns, 
                    v_config.source_table_name, p_source_dblink
                );
                
                IF v_has_pk THEN
                    v_time_cond := 's.' || p_timestamp_col || ' > t.' || p_timestamp_col;
                    v_merge_sql := 'MERGE INTO ' || p_table_name || ' t ' ||
                                  'USING (' || v_paged_query || ') s ' ||
                                  'ON (' || v_key_cond || ') ' ||
                                  'WHEN MATCHED  THEN ' ||
                                  '  UPDATE SET ' || v_update_list  ||
                                --  '  t.' || p_timestamp_col || ' = s.' || p_timestamp_col || ' ' ||
                                  '  WHERE (' || v_time_cond || ')' ||
                                  'WHEN NOT MATCHED THEN ' ||
                                  '  INSERT (' || p_col_list || ') ' ||
                                  '  VALUES ('|| v_select_cols|| ')';
                                  --'  VALUES (s.' || REPLACE(p_col_list, ',', ', s.') || ')';
                ELSE
                    v_merge_sql := 'INSERT INTO ' || p_table_name || ' (' || p_col_list || ') ' ||
                                  'SELECT ' || v_select_cols || ' FROM (' || v_paged_query || ') s ' ||
                                  'WHERE NOT EXISTS (SELECT 1 FROM ' || p_table_name || ' t ' ||
                                  '  WHERE t.' || p_timestamp_col || ' = s.' || p_timestamp_col || ')';
                END IF;
                
                log_sql(v_merge_sql);
                EXECUTE IMMEDIATE v_merge_sql;
                log_sql_result(p_operation || 'BATCH' || i, SQL%ROWCOUNT, SYSTIMESTAMP);
                log_sync_history(p_table_name, 'INCREMENTAL', p_operation || 'BATCH' || i, 
                                SQL%ROWCOUNT, 'SUCCESS', NULL, p_batch_id);
            END;
            
            -- 调整批次范围
            v_start_row := v_start_row + p_batch_size;
            v_end_row := v_end_row + p_batch_size;
            COMMIT;
        END LOOP;
        
        log_msg(' 分批处理完成，总耗时: ' || TO_CHAR((SYSTIMESTAMP - v_start_time) * 24 * 60 * 60, '999999.999') || ' 秒');
    EXCEPTION
        WHEN OTHERS THEN
            log_error(' 分批处理失败: ' || SQLERRM);
            log_sync_history(p_table_name, 'INCREMENTAL', 'BATCH_ERROR', 0, 'ERROR', SQLERRM, p_batch_id);
            RAISE;
    END;
    
    -- 批量删除过程（支持软删除和硬删除）
    PROCEDURE batch_delete(
        p_table_name    VARCHAR2,
        p_source_table  VARCHAR2,
        p_source_dblink VARCHAR2,
        p_source_query  VARCHAR2,
        p_key_columns   VARCHAR2,
        p_batch_id      NUMBER
    ) IS
        v_config config_rec;
        v_sql VARCHAR2(32767);
        v_condition VARCHAR2(4000);
        v_key_list VARCHAR2(4000) := '';
        v_value_list VARCHAR2(4000) := '';
        v_delimiter CONSTANT VARCHAR2(1) := ',';
        v_start_time TIMESTAMP := SYSTIMESTAMP;
        v_deleted_rows NUMBER := 0;
    BEGIN
        v_config := get_config(p_table_name);
        v_condition := build_key_condition(
            p_key_columns, 't', 's', p_table_name, 
            p_source_table, p_source_dblink
        );
        
        log_msg(' 删除条件: ' || v_condition);
        
        -- 处理软删除列存在性（支持 DB Link）
        IF v_config.soft_delete_col IS NOT NULL AND TRIM(v_config.soft_delete_col) IS NOT NULL THEN
            IF NOT is_column_exists(p_source_table, v_config.soft_delete_col, p_source_dblink) THEN
                log_msg(' 警告：软删除列 "' || v_config.soft_delete_col || 
                        '" 在源表中不存在，使用硬删除');
               -- v_config.soft_delete_col := NULL;
            END IF;
        END IF;
        
        -- 构建键值对用于删除历史记录
        DECLARE
            v_idx PLS_INTEGER := 1;
            v_key VARCHAR2(100);
            v_pos PLS_INTEGER;
        BEGIN
            WHILE v_idx <= LENGTH(p_key_columns) LOOP
                v_pos := INSTR(p_key_columns, v_delimiter, v_idx);
                IF v_pos > 0 THEN
                    v_key := TRIM(SUBSTR(p_key_columns, v_idx, v_pos - v_idx));
                    v_idx := v_pos + 1;
                ELSE
                    v_key := TRIM(SUBSTR(p_key_columns, v_idx));
                    v_idx := LENGTH(p_key_columns) + 1;
                END IF;
                v_key_list := v_key_list || '''' || v_key || '''';
                v_value_list := v_value_list || 't.' || v_key;
                IF v_pos > 0 THEN
                    v_key_list := v_key_list || ', ';
                    v_value_list := v_value_list || ', ';
                END IF;
            END LOOP;
        END;
        
        -- 软删除处理
        IF v_config.soft_delete_col IS NOT NULL AND TRIM(v_config.soft_delete_col) IS NOT NULL THEN
            log_msg(' 执行软删除，标志列: ' || v_config.soft_delete_col);
            
            -- 记录删除历史
            v_sql := 'INSERT INTO sync_deleted_history (
                deleted_id, table_name, key_values, delete_time, sync_batch_id
            ) SELECT
                sync_deleted_history_seq.NEXTVAL, ''' || p_table_name || ''',
                JSON_OBJECT (' || v_key_list || ' VALUE ' || v_value_list || '),
                SYSTIMESTAMP, ' || p_batch_id || '
            FROM ' || p_table_name || ' t
            WHERE NOT EXISTS (SELECT 1 FROM (' || p_source_query || ') s
            WHERE ' || v_condition || ')';
            
            log_sql(v_sql);
            EXECUTE IMMEDIATE v_sql;
            
            -- 执行软删除（更新标志列）
            v_sql := 'UPDATE ' || p_table_name || ' t
            SET ' || v_config.soft_delete_col || ' = 1 , DELETE_TIME =  SYSTIMESTAMP
            WHERE NOT EXISTS (SELECT 1 FROM (' || p_source_query || ') s
            WHERE ' || v_condition || ')';
            
            log_sql(v_sql);
            EXECUTE IMMEDIATE v_sql;
            v_deleted_rows := SQL%ROWCOUNT;
            log_sql_result(' 软删除 ', v_deleted_rows, v_start_time);
            
        -- 硬删除处理
        ELSE
            log_msg(' 执行硬删除 ');
            
            -- 记录删除历史
            v_sql := 'INSERT INTO sync_deleted_history (
                deleted_id, table_name, key_values, delete_time, sync_batch_id
            ) SELECT
                sync_deleted_history_seq.NEXTVAL, ''' || p_table_name || ''',
                JSON_OBJECT (' || v_key_list || ' VALUE ' || v_value_list || '),
                SYSTIMESTAMP, ' || p_batch_id || '
            FROM ' || p_table_name || ' t
            WHERE NOT EXISTS (SELECT 1 FROM (' || p_source_query || ') s
            WHERE ' || v_condition || ')';
            
            log_sql(v_sql);
            EXECUTE IMMEDIATE v_sql;
            
            -- 执行硬删除
            v_sql := 'DELETE FROM ' || p_table_name || ' t
            WHERE NOT EXISTS (SELECT 1 FROM (' || p_source_query || ') s
            WHERE ' || v_condition || ')';
            
            log_sql(v_sql);
            EXECUTE IMMEDIATE v_sql;
            v_deleted_rows := SQL%ROWCOUNT;
            log_sql_result(' 硬删除 ', v_deleted_rows, v_start_time);
        END IF;
        
        -- 记录同步历史
        log_sync_history(p_table_name, 'INCREMENTAL', 'DELETE', v_deleted_rows, 'SUCCESS', NULL, p_batch_id);
    EXCEPTION
        WHEN OTHERS THEN
            log_error(' 删除失败: ' || SQLERRM);
            log_sync_history(p_table_name, 'INCREMENTAL', 'DELETE', 0, 'ERROR', SQLERRM, p_batch_id);
            RAISE;
    END;
    
    -- 主同步过程
    PROCEDURE sync_table(p_table_name IN VARCHAR2) IS
        v_config config_rec;
        v_start_time TIMESTAMP := SYSTIMESTAMP;
        v_error_message VARCHAR2(4000);
        v_batch_id NUMBER;
        v_source_query VARCHAR2(32767);
        v_full_query VARCHAR2(32767);
        v_last_sync_time_str VARCHAR2(50);
        v_col_list VARCHAR2(4000);
        v_batch_size NUMBER;
    BEGIN
        log_msg(' 开始同步表: ' || p_table_name);
        
        v_config := get_config(p_table_name);
        v_batch_id := sync_history_seq.NEXTVAL;
        v_batch_size := NVL(v_config.batch_size, 10000);
        
        log_sync_history(p_table_name,
                         CASE WHEN v_config.last_sync_time IS NULL THEN 'FULL' ELSE 'INCREMENTAL' END,
                         'START', 0, 'IN_PROGRESS', NULL, v_batch_id);
        
        -- 构建全量查询
        v_full_query := NVL(v_config.full_sql, 
                           'SELECT * FROM ' || v_config.source_table_name || '@' || v_config.source_dblink);
        
        v_col_list := build_column_list(p_table_name, 
                                      v_config.key_columns, 
                                      v_config.extra_columns, 
                                      v_config.source_table_name, 
                                      v_config.source_dblink);

        BEGIN
            -- 首次全量同步（保留数据）
	        IF v_config.last_sync_time IS NULL THEN
	            log_msg(' 执行首次全量同步（保留数据）');
	            
	            -- 全量插入 / 更新
	            DECLARE
	                v_merge_sql VARCHAR2(32767);
	                v_key_cond VARCHAR2(4000);
	                v_update_list VARCHAR2(32767);
	                v_time_cond VARCHAR2(1000);
	                v_select_cols VARCHAR2(32767);
	            BEGIN
	                v_key_cond := build_key_condition(
	                    v_config.key_columns, 't', 's', p_table_name, 
	                    v_config.source_table_name, v_config.source_dblink
	                );
	                
	                v_update_list := build_update_list(
	                    p_table_name, v_config.key_columns, v_config.extra_columns, 
	                    v_config.source_table_name, v_config.source_dblink
	                );
	                
	                -- 生成带默认值的SELECT列列表
	                v_select_cols := generate_select_columns(v_col_list, v_config.source_table_name, v_config.source_dblink);
	                
	                IF v_config.has_primary_key THEN
	                    v_time_cond := 's.' || v_config.timestamp_column || ' > t.' || v_config.timestamp_column;
	                    v_merge_sql := 'MERGE INTO ' || p_table_name || ' t ' ||
	                                  'USING (' || v_full_query || ') s ' ||
	                                  'ON (' || v_key_cond || ') ' ||
	                                  'WHEN MATCHED  THEN ' ||
	                                  '  UPDATE SET ' || v_update_list || --', ' ||
	                                 -- '  t.' || v_config.timestamp_column || ' = s.' || v_config.timestamp_column || ' ' ||
	                                  '  WHERE (' || v_time_cond || ')' ||
	                                  'WHEN NOT MATCHED THEN ' ||
	                                  '  INSERT (' || v_col_list || ') ' ||
	                                  '  VALUES (' || v_select_cols || ')';
	                ELSE
	                    v_merge_sql := 'INSERT INTO ' || p_table_name || ' (' || v_col_list || ') ' ||
	                                  'SELECT ' || v_select_cols || ' FROM (' || v_full_query || ') s ' ||
	                                  'WHERE NOT EXISTS (SELECT 1 FROM ' || p_table_name || ' t ' ||
	                                  '  WHERE t.' || v_config.timestamp_column || ' = s.' || v_config.timestamp_column || ')';
	                END IF;
	                
	                log_sql(v_merge_sql);
	                EXECUTE IMMEDIATE v_merge_sql;
	                log_sql_result(' 全量插入/更新 ', SQL%ROWCOUNT, SYSTIMESTAMP);
	                log_sync_history(p_table_name, 'FULL', 'INSERT', SQL%ROWCOUNT, 'SUCCESS', NULL, v_batch_id);
	            END;
	            
	            -- 处理删除
	            batch_delete(p_table_name, 
	                       v_config.source_table_name, 
	                       v_config.source_dblink, 
	                       v_full_query, 
	                       v_config.key_columns, 
	                       v_batch_id);
                           
            -- 增量同步
            ELSE
                log_msg(' 执行增量同步，最后同步时间: ' || 
                        TO_CHAR(v_config.last_sync_time, 'YYYY-MM-DD HH24:MI:SS'));
                
                v_last_sync_time_str := TO_CHAR(v_config.last_sync_time, 'YYYY-MM-DD HH24:MI:SS.FF9');
                
                -- 构建增量查询
                v_source_query := 'SELECT * FROM (' || v_full_query || ') ' ||
                                 'WHERE ' || v_config.timestamp_column || ' > ' ||
                                 'TO_TIMESTAMP(''' || v_last_sync_time_str || ''', ''YYYY-MM-DD HH24:MI:SS.FF9'')';
                
                log_msg(' 增量查询: ' || v_source_query);
                
                -- 增量插入 / 更新（分批处理）
                IF v_batch_size > 0 THEN
                    process_batch(p_table_name, 
                               v_source_query, 
                               v_batch_size, 
                               v_config.key_columns, 
                               v_config.timestamp_column, 
                               v_col_list, 
                               v_batch_id, 
                               'INCREMENTAL_UPDATE', 
                               v_config.source_dblink);
                ELSE
                    DECLARE
                        v_merge_sql VARCHAR2(32767);
                        v_key_cond VARCHAR2(4000);
                        v_update_list VARCHAR2(32767);
                        v_time_cond VARCHAR2(1000);
                    BEGIN
                        v_key_cond := build_key_condition(
                            v_config.key_columns, 't', 's', p_table_name, 
                            v_config.source_table_name, v_config.source_dblink
                        );
                        
                        v_update_list := build_update_list(
                            p_table_name, v_config.key_columns, v_config.extra_columns, 
                            v_config.source_table_name, v_config.source_dblink
                        );
                        
                        IF v_config.has_primary_key THEN
                            v_time_cond := 's.' || v_config.timestamp_column || ' > t.' || v_config.timestamp_column;
                            v_merge_sql := 'MERGE INTO ' || p_table_name || ' t ' ||
                                          'USING (' || v_source_query || ') s ' ||
                                          'ON (' || v_key_cond || ') ' ||
                                          'WHEN MATCHED AND (' || v_time_cond || ') THEN ' ||
                                          '  UPDATE SET ' || v_update_list || ', ' ||
                                          '  t.' || v_config.timestamp_column || ' = s.' || v_config.timestamp_column || ' ' ||
                                          'WHEN NOT MATCHED THEN ' ||
                                          '  INSERT (' || v_col_list || ') ' ||
                                          '  VALUES (s.' || REPLACE(v_col_list, ',', ', s.') || ')';
                        ELSE
                            v_merge_sql := 'INSERT INTO ' || p_table_name || ' (' || v_col_list || ') ' ||
                                          'SELECT ' || v_col_list || ' FROM (' || v_source_query || ') s ' ||
                                          'WHERE NOT EXISTS (SELECT 1 FROM ' || p_table_name || ' t ' ||
                                          '  WHERE t.' || v_config.timestamp_column || ' = s.' || v_config.timestamp_column || ')';
                        END IF;
                        
                        log_sql(v_merge_sql);
                        EXECUTE IMMEDIATE v_merge_sql;
                        log_sql_result(' 增量更新 ', SQL%ROWCOUNT, SYSTIMESTAMP);
                        log_sync_history(p_table_name, 'INCREMENTAL', 'UPDATE', SQL%ROWCOUNT, 'SUCCESS', NULL, v_batch_id);
                    END;
                END IF;
                
                -- 增量删除
                batch_delete(p_table_name, 
                           v_config.source_table_name, 
                           v_config.source_dblink, 
                           v_full_query, 
                           v_config.key_columns, 
                           v_batch_id);
            END IF;
            
            -- 更新同步时间
            UPDATE sync_config
            SET last_sync_time = SYSTIMESTAMP, last_sync_batch = v_batch_id
            WHERE table_name = p_table_name;
            
            COMMIT;
            
            log_sync_history(p_table_name,
                             CASE WHEN v_config.last_sync_time IS NULL THEN 'FULL' ELSE 'INCREMENTAL' END,
                             'COMPLETE', 0, 'SUCCESS', NULL, v_batch_id);
            
            log_msg(' 表同步完成，总耗时: ' || TO_CHAR((SYSTIMESTAMP - v_start_time) * 24 * 60 * 60, '999999.999') || ' 秒');
            
        EXCEPTION
            WHEN OTHERS THEN
                v_error_message := SQLERRM;
                log_error(' 同步异常: ' || v_error_message);
                log_sync_history(p_table_name,
                                 CASE WHEN v_config.last_sync_time IS NULL THEN 'FULL' ELSE 'INCREMENTAL' END,
                                 'ERROR', 0, 'ERROR', v_error_message, v_batch_id);
                RAISE;
        END;
    END;
    
    -- 批量同步过程
    PROCEDURE sync_tables IS
        CURSOR c_tables IS
            SELECT table_name FROM sync_config WHERE enable_flag = 'Y';
        v_start_time TIMESTAMP := SYSTIMESTAMP;
        v_table_count NUMBER := 0;
        v_success_count NUMBER := 0;
        v_failed_count NUMBER := 0;
        v_batch_id NUMBER;
        v_global_sql VARCHAR2(32767);
    BEGIN
        log_msg(' 开始批量同步所有表 ');
        
        -- 记录全局同步开始
        v_global_sql := 'INSERT INTO sync_global_history (
            history_id, start_time, status
        ) VALUES (
            sync_global_history_seq.NEXTVAL, SYSTIMESTAMP, ''IN_PROGRESS''
        ) RETURNING history_id INTO :1';
        EXECUTE IMMEDIATE v_global_sql USING OUT v_batch_id;
        COMMIT;
        log_msg(' 全局批次 ID: ' || v_batch_id);
        
        -- 同步每个表前清空缓存
        g_column_cache.DELETE;
        
        -- 同步每个表
        FOR r_table IN c_tables LOOP
            v_table_count := v_table_count + 1;
            BEGIN
                log_msg(' 同步表 [' || v_table_count || ']: ' || r_table.table_name);
                sync_table(r_table.table_name);
                v_success_count := v_success_count + 1;
                log_msg(' 表同步成功: ' || r_table.table_name);
            EXCEPTION
                WHEN OTHERS THEN
                    log_error(' 表同步失败: ' || r_table.table_name || ', 错误: ' || SQLERRM);
                    v_failed_count := v_failed_count + 1;
            END;
        END LOOP;
        
        -- 更新全局状态
        v_global_sql := 'UPDATE sync_global_history
            SET end_time = SYSTIMESTAMP,
                total_tables = :1,
                success_tables = :2,
                failed_tables = :3,
                status = CASE WHEN :3 = 0 THEN ''SUCCESS'' ELSE ''PARTIAL_SUCCESS'' END
            WHERE history_id = :4';
        EXECUTE IMMEDIATE v_global_sql USING v_table_count, v_success_count, v_failed_count, v_batch_id;
        COMMIT;
        
        log_msg(' 批量同步完成：总表数 = ' || v_table_count || ', 成功 = ' ||
                v_success_count || ', 失败 = ' || v_failed_count || ', 状态 = ' || 
                CASE WHEN v_failed_count = 0 THEN 'SUCCESS' ELSE 'PARTIAL_SUCCESS' END ||
                ', 总耗时: ' || TO_CHAR((SYSTIMESTAMP - v_start_time) * 24 * 60 * 60, '999999.999') || ' 秒');
    END;
    
    -- 添加新同步表
    PROCEDURE add_sync_table(
        p_table_name        VARCHAR2,
        p_source_table_name VARCHAR2,
        p_source_dblink     VARCHAR2,
        p_key_columns       VARCHAR2,
        p_timestamp_column  VARCHAR2,
        p_full_sql          CLOB,
        p_batch_size        NUMBER DEFAULT 10000,
        p_parallel_degree   NUMBER DEFAULT 4,
        p_soft_delete_col   VARCHAR2 DEFAULT NULL,
        p_extra_columns     VARCHAR2 DEFAULT NULL
    ) IS
        v_sql VARCHAR2(4000);
    BEGIN
        log_msg(' 添加同步表: ' || p_table_name);
        
        -- 校验额外列和软删除列的存在性（支持 DB Link）
        IF p_extra_columns IS NOT NULL AND TRIM(p_extra_columns) != '' THEN
            DECLARE
                v_extra_col VARCHAR2(100);
                v_extra_pos PLS_INTEGER := 1;
                v_extra_end PLS_INTEGER;
            BEGIN
                LOOP
                    v_extra_end := INSTR(p_extra_columns, ',', v_extra_pos);
                    IF v_extra_end > 0 THEN
                        v_extra_col := TRIM(SUBSTR(p_extra_columns, v_extra_pos, v_extra_end - v_extra_pos));
                        v_extra_pos := v_extra_end + 1;
                    ELSE
                        v_extra_col := TRIM(SUBSTR(p_extra_columns, v_extra_pos));
                        v_extra_pos := LENGTH(p_extra_columns) + 1;
                    END IF;
                    IF NOT is_column_exists(p_source_table_name, v_extra_col, p_source_dblink) THEN
                        RAISE_APPLICATION_ERROR(-20004, 
                                                ' 源表 ' || p_source_table_name || ' 缺少额外列 ' || v_extra_col);
                    END IF;
                END LOOP;
            END;
        END IF;
        
        -- 校验软删除列的存在性（支持 DB Link）
        IF p_soft_delete_col IS NOT NULL AND TRIM(p_soft_delete_col) != '' THEN
            IF NOT is_column_exists(p_source_table_name, p_soft_delete_col, p_source_dblink) THEN
                RAISE_APPLICATION_ERROR(-20004, 
                                        ' 源表 ' || p_source_table_name || ' 缺少软删除列 ' || p_soft_delete_col);
            END IF;
        END IF;
        
        v_sql := 'INSERT INTO sync_config (
            table_name, source_table_name, source_dblink, key_columns, timestamp_column,
            full_sql, batch_size, parallel_degree, soft_delete_col, extra_columns,
            enable_flag, created_by, created_time, updated_by, updated_time
        ) VALUES (
            :1, :2, :3, :4, :5,
            :6, :7, :8, :9, :10,
            ''Y'', :11, SYSTIMESTAMP, :11, SYSTIMESTAMP
        )';
        EXECUTE IMMEDIATE v_sql USING p_table_name, p_source_table_name, p_source_dblink, 
                            p_key_columns, p_timestamp_column, p_full_sql, p_batch_size, 
                            p_parallel_degree, p_soft_delete_col, p_extra_columns, USER;
        COMMIT;
        log_msg(' 表添加成功: ' || p_table_name);
    END;
    
    -- 启用/禁用同步表
    PROCEDURE toggle_sync_table(p_table_name VARCHAR2, p_enable CHAR DEFAULT 'Y') IS
        v_sql VARCHAR2(2000);
    BEGIN
        log_msg(' 切换表状态: ' || p_table_name || ' 启用 = ' || p_enable);
        v_sql := 'UPDATE sync_config
            SET enable_flag = :1, updated_by = :2, updated_time = SYSTIMESTAMP
            WHERE table_name = :3';
        EXECUTE IMMEDIATE v_sql USING p_enable, USER, p_table_name;
        COMMIT;
        log_msg(' 表状态已更新 ');
    END;
    
    -- 更新全量SQL
    PROCEDURE update_full_sql(p_table_name VARCHAR2, p_full_sql CLOB) IS
        v_sql VARCHAR2(2000);
    BEGIN
        log_msg(' 更新全量 SQL: ' || p_table_name);
        v_sql := 'UPDATE sync_config
            SET full_sql = :1, last_sync_time = NULL, updated_by = :2, updated_time = SYSTIMESTAMP
            WHERE table_name = :3';
        EXECUTE IMMEDIATE v_sql USING p_full_sql, USER, p_table_name;
        COMMIT;
        log_msg(' 全量 SQL 更新成功 ');
    END;
    
    -- 手动执行全量同步
    PROCEDURE full_sync_table(p_table_name IN VARCHAR2) IS
    BEGIN
        log_msg(' 触发全量同步: ' || p_table_name);
        UPDATE sync_config SET last_sync_time = NULL WHERE table_name = p_table_name;
        COMMIT;
        sync_table(p_table_name);
        log_msg(' 全量同步完成 ');
    END;
    
    -- 查看同步状态
    FUNCTION get_sync_status(p_table_name VARCHAR2) RETURN VARCHAR2 IS
        v_status VARCHAR2(20);
    BEGIN
        SELECT status
        INTO v_status
        FROM sync_history
        WHERE table_name = p_table_name
        AND operation = 'COMPLETE'
        ORDER BY history_id DESC
        FETCH FIRST 1 ROW ONLY;
        RETURN v_status;
    EXCEPTION
        WHEN NO_DATA_FOUND THEN
            RETURN 'NEVER SYNCED';
    END;
    
    -- 获取最后同步时间
    FUNCTION get_last_sync_time(p_table_name VARCHAR2) RETURN TIMESTAMP IS
        v_time TIMESTAMP;
    BEGIN
        SELECT last_sync_time
        INTO v_time
        FROM sync_config
        WHERE table_name = p_table_name;
        RETURN v_time;
    EXCEPTION
        WHEN NO_DATA_FOUND THEN
            RETURN NULL;
    END;
    
    -- 获取同步统计信息
    FUNCTION get_sync_stats(p_table_name VARCHAR2) RETURN NUMBER IS
        v_count NUMBER;
    BEGIN
        SELECT COUNT(*)
        INTO v_count
        FROM sync_history
        WHERE table_name = p_table_name
        AND status = 'SUCCESS';
        RETURN v_count;
    EXCEPTION
        WHEN NO_DATA_FOUND THEN
            RETURN 0;
    END;
END table_sync_pkg;